package com.wolffy.flink.transfrom;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


public class RichMapFunctionDemo {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> source = env.socketTextStream("hadoop102", 7777);
        SingleOutputStreamOperator<Integer> map = source.map(new RichMapFunction<String, Integer>() {
            @Override
            public void open(Configuration parameters) throws Exception {
                //open(): 每个子任务，在启动时，调用一次
                super.open(parameters);
                System.out.println("子任务编号=" + getRuntimeContext().getIndexOfThisSubtask() + ",子任务名称=" + getRuntimeContext().getTaskNameWithSubtasks() + ",调用open()");
            }
            @Override
            public void close() throws Exception {
                //close():每个子任务，在结束时，调用一次
                super.close();
                System.out.println(
                        "子任务编号=" + getRuntimeContext().getIndexOfThisSubtask() + "，子任务名称=" + getRuntimeContext().getTaskNameWithSubtasks() + ",调用close()");
            }
            @Override
            public Integer map(String value) throws Exception {
                return Integer.parseInt(value) + 1;
            }
        });

        map.print();

        env.execute();
    }
}
